package cn.ilikes.tools.nosql.jpa.clients.indexs.index.hadoops;

import java.util.Collections;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooDefs.Perms;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

public class HadoopZooKeepLockUnilts {

	private static byte[] createLock(String lock_Id, LockType lockType) {
		String lock = lock_Id + ":" + lockType.toString();
		return lock.getBytes();
	}

	private static boolean isMyLock(String lock, String myId) {

		boolean status = false;
		String[] x = lock.split(":");
		if (x[0].trim().equals(myId)) {

			status = true;
		} else {

			status = false;
		}
		return status;
	}

	private static LockType lockType(String lock, String myId) {

		String[] x = lock.split(":");
		LockType lockType = LockType.valueOf(x[1]);
		return lockType;
	}

	private enum LockType {
		lock, unlock
	}

	public static boolean obtainLock(ZooKeeper zooKeeper, String path, String lock_Id) throws Exception {
		boolean status = false;
		Stat stat = zooKeeper.exists(path, true);
		
		if (stat == null) {
			zooKeeper.create(path, createLock(lock_Id, LockType.lock), Collections.singletonList(new ACL(Perms.ALL, Ids.ANYONE_ID_UNSAFE)), CreateMode.EPHEMERAL);
			status = true;
		} else {
			String lock = new String(zooKeeper.getData(path, true, stat));
			LockType locktype = lockType(lock, lock_Id);
			if (locktype.equals(LockType.unlock)) {
				zooKeeper.setData(path, createLock(lock_Id, LockType.lock), stat.getVersion());
				status = true;
			} else {
				status = false;
			}

		}
		return status;
	}

	public static boolean isLocked(ZooKeeper zooKeeper, String path, String lock_Id) throws Exception {
		Stat stat = zooKeeper.exists(path, true);
		if (stat == null) {
			return false;
		} else {
			String lock = new String(zooKeeper.getData(path, true, stat));
			LockType locktype = lockType(lock, lock_Id);
			switch (locktype) {
			case unlock:
				return false;
			case lock:
				return true;
			default:
				throw new Exception("get lock type error :" + lock);

			}
		}

	}

	public static void release(ZooKeeper zooKeeper, String path, String lock_Id) throws Exception {
		Stat stat = zooKeeper.exists(path, true);
		if (stat == null) {
			zooKeeper.create(path, createLock(lock_Id, LockType.unlock), Collections.singletonList(new ACL(Perms.ALL, Ids.ANYONE_ID_UNSAFE)), CreateMode.EPHEMERAL);
		} else {
			String lock = new String(zooKeeper.getData(path, true, stat));
			boolean mylock = isMyLock(lock, lock_Id);
			if (mylock) {
				zooKeeper.setData(path, createLock(lock_Id, LockType.unlock), stat.getVersion());
			}
		}
	}

	public static void remove(ZooKeeper zooKeeper, String path, String lock_Id) throws Exception {
		Stat stat = zooKeeper.exists(path, true);
		if (stat == null) {
			zooKeeper.create(path, LockType.unlock.toString().getBytes(), Collections.singletonList(new ACL(Perms.ALL, Ids.ANYONE_ID_UNSAFE)), CreateMode.EPHEMERAL);
		} else {
			String lock = new String(zooKeeper.getData(path, true, stat));
			boolean mylock = isMyLock(lock, lock_Id);
			if (mylock)
				zooKeeper.setData(path, createLock(lock_Id, LockType.unlock), stat.getVersion());
		}

	}
}
